package com.atguigu.gmall.realtime.app.dwd.db;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.BaseDbTableProcessFunction;
import com.atguigu.gmall.realtime.bean.BaseDbTableProcess;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;

/**
 * @author Felix
 * @date 2023/8/26
 * 事实表动态分流处理
 * 需要启动的进程
 *      zk、kafka、maxwell、BaseDbApp
 * 代码开发流程
 *      准备环境
 *      检查点设置
 *      从kafka中读取数据
 *      类型转换以及ETL
 *      使用FlinkCDC读取配置表数据
 *      广播---broadcast
 *      主流和广播流进行关联---connect
 *      对关联后的数据进行处理---process
 *      class BaseTableProcessFunction extends BroadcastProcessFunction{
 *          open:预加载配置信息到程序中
 *          processElement:处理主流数据
 *              根据表名和操作类型作为key，到广播状态以及configMap中获取对应的配置信息
 *              如果配置信息不为空，说明是需要动态分流处理的事实数据，传递到下游
 *                  在向下游传递数据前，过滤掉不需要传递的属性、补充了输出的目的地、补充了事件时间
 *          processBroadcastElement:处理广播流数据
 *              op="d"
 *                  将配置信息从广播状态以及configMap中删除掉
 *              op!="d"
 *                  将配置信息放到广播状态以及configMap中
 *      }
 *     将流中数据写到kafka的不同的主题中
 *          MyKafkaUtil---<T>KafkaSink<T> getKafkaSinkBySchema(序列化)
 */
public class BaseDbApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        /*//TODO 2.检查点相关的设置
        //2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置job取消后检查点是否保留
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置两个检查点最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
        //2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        //2.7 设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");
        */
        //TODO 3.从kafka主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "topic_db";
        String groupId = "base_db_app_group";
        //3.2 创建消费者对象
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 消费数据 封装为流
        DataStreamSource<String> kafkaStrDS
            = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        // kafkaStrDS.print(">>>");

        //TODO 4.对流中的数据进行类型转换并进行ETL
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
            new ProcessFunction<String, JSONObject>() {
                @Override
                public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                    try {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        String type = jsonObj.getString("type");
                        if (!type.startsWith("bootstrap-")) {
                            out.collect(jsonObj);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }

                }
            }
        );
        // jsonObjDS.print(">>>");
        //TODO 5.使用FlinkCDC读取配置表中的配置信息
        //5.1 创建MySource
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
            .hostname("hadoop102")
            .port(3306)
            .databaseList("gmall0315_config") // set captured database
            .tableList("gmall0315_config.table_process_dwd") // set captured table
            .username("root")
            .password("123456")
            .startupOptions(StartupOptions.initial())
            .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
            .build();
        //5.2 封装为流
        DataStreamSource<String> mysqlStrDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql_source");
        // mysqlStrDS.print(">>>");

        //TODO 6.对读取的配置信息进行广播
        MapStateDescriptor<String, BaseDbTableProcess> mapStateDescriptor
            = new MapStateDescriptor<String, BaseDbTableProcess>("mapStateDescriptor",String.class,BaseDbTableProcess.class);
        BroadcastStream<String> broadcastDS = mysqlStrDS.broadcast(mapStateDescriptor);

        //TODO 7.关联主流业务数据以及广播流中的配置数据
        BroadcastConnectedStream<JSONObject, String> connectDS = jsonObjDS.connect(broadcastDS);

        //TODO 8.对关联后的数据进行处理
        SingleOutputStreamOperator<JSONObject> realDS = connectDS.process(
            new BaseDbTableProcessFunction(mapStateDescriptor)
        );
        //TODO 9.将需要动态分流处理的事实数据写到kafka的不同主题中
        realDS.print(">>>");
        realDS.sinkTo(
            MyKafkaUtil.getKafkaSinkBySchema(new KafkaRecordSerializationSchema<JSONObject>() {
                @Nullable
                @Override
                public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObj, KafkaSinkContext context, Long timestamp) {
                    String sinkTopic = jsonObj.getString("sink_table");
                    jsonObj.remove("sink_table");

                    return new ProducerRecord<byte[], byte[]>(sinkTopic,jsonObj.toJSONString().getBytes());
                }
            })
        );
        env.execute();
    }
}
